package com.flinksql.test;

import com.flinksql.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;


/**
 * @author: Lin
 * @create: 2021-06-16 10:21
 * @description: 声明时间语义，基于时间时间，table写法
 **/
public class FlinkTableAPI_Test11_EvenTime {
    public static void main(String[] args) throws Exception {
        //1.建立环境，测试不设置CK
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2.获取数据，并指定watermark
        WatermarkStrategy<WaterSensor> wms = WatermarkStrategy
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs();
            }
        });

        SingleOutputStreamOperator<WaterSensor> mapDS = env.readTextFile("input/sensor.txt")
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                }).assignTimestampsAndWatermarks(wms);

        //3.在table中指定时间属性
        Table table = tableEnv.fromDataStream(mapDS, $("id"), $("ts"), $("vc")
                , $("rt").rowtime());

        table.printSchema();


    }
}
